package com.nebula.rocketmq.controller;

import com.nebula.rocketmq.exception.RocketMQException;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.annotation.CacheEvict;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


/**
 * @Author: llq
 * @Date: 2019/9/18
 * 当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。
 * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息，然后再回调用户Listener方法
 *
 */
@RestController
@RequestMapping("/rocketmq")
public class RockerMQConsumer {
    @Value("${rocketmq.consumer.groupName}")
    private String groupName;

    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;

    @Value("${rocketmq.consumer.consumeThreadMin}")
    private int consumeThreadMin;

    @Value("${rocketmq.consumer.consumeThreadMax}")
    private int consumeThreadMax;

    @Value("${rocketmq.consumer.topic}")
    private String topic;

    @Value("${rocketmq.consumer.tag}")
    private String tag;

    @GetMapping("/get")
    //清除1条缓存数据
    @CacheEvict(value = "cachemq",allEntries = false,beforeInvocation = false)
    public void getRocketMQConsumer() throws RocketMQException {
        if (StringUtils.isBlank(groupName)){
            throw new RocketMQException("groupName is null !!!");
        }
        if (StringUtils.isBlank(namesrvAddr)){
            throw new RocketMQException("namesrvAddr is null !!!");
        }
        if (StringUtils.isBlank(topic)){
            throw new RocketMQException("topic is null !!!");
        }
        if (StringUtils.isBlank(tag)){
            throw new RocketMQException("tag is null !!!");
        }
        /**
         * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例
         * 注意：groupName需要由应用来保证唯一
         */
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);

        //创建监听器
//        MessageListener messageListener = new MessageListener();
        //推模式，消费者将自己注册到监听器MessageListener里
//1        consumer.registerMessageListener(messageListener);

        try {
            /*// 另外一种获取消息的方法 :从指定topic中拉取所有消息队列 ........
            Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("order-topic")
            .....................;*/
            /**
             * 1.订阅主题来消费,主要用到subscribe()
             *             消费者订阅topic主题下的tag为tag的消息
             */
            consumer.subscribe(topic,tag);
            /**
             * 订阅指定topic下所有消息<br>
             * 注意：一个consumer对象可以订阅多个topic
             */
            //consumer.subscribe("Topic2", "*");

            //2.设置Push的消费策略，用到setConsumeFromWhere()  没写,使用的默认消费策略ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
            //设置是从队列头部还是尾部开始消费，如果非第一次启动，那么按照上次消费的位置继续消费
            // consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            // 开启内部类实现监听
            //取到消息后,唤醒MessageListenerConcurrently的consumeMessage()来消费
            //参数list是一个消息列表
            //参数context用于消费的事务控制，可以用于设置是否自动提交，消息队列和当前队列的暂停时间
            consumer.registerMessageListener((MessageListenerConcurrently) (list,context) -> {
                //具体拿到消息的逻辑
                //list就是结果
                try {
                    for (MessageExt messageExt : list) {
                        //反序列化
                        String str = new String(messageExt.getBody());
                        //输出消息内容
                        System.err.println("消费消息: " + str);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    //当监听是并发消费MessageListenerConcurrently，所以使用ConsumeConcurrentlyStatus并发返回值
                    //当监听是有序消费MessageListenerOrderly，使用ConsumeOrderlyStatus顺序返回值
                    //并发消费返回值 ：失败稍后再试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //并发消费返回值 ：消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            /*// 开启内部类实现监听
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    return DefaultConsumerConfigure.this.dealBody(msgs);
                }
            });*/

            //Consumer对象在使用之前必须要调用start初始化，初始化一次即可
            consumer.start();
        }catch (MQClientException e){
            throw new RocketMQException(e);
        }
    }
}
